package com.tjf.myjetpackdemo.RxObserverModel;

/**
 * @author: tjf
 * @date: 2023-06-07
 * @desc： 多个链式 抽象的被观察者 和抽象的观察者 之间的绑定处理
 */
public class ObservableMap<T, U> extends AbstractObservableWithUpstream<T, U> {


    final Function<T, U> function;

    protected ObservableMap(ObservableSource<T> source, Function<T, U> function) {
        super(source);
        this.function = function;
    }

    @Override
    protected void subscribeActual(Observer<U> observer) {
        source.subscribe(new MapObserver(observer, function));
    }

    static final class MapObserver<T, U> extends BaiseFuseableOberver<T, U> {
        final Function<T, U> mapper;

        public MapObserver(Observer<U> actual, Function<T, U> mapper) {
            super(actual);
            this.mapper = mapper;
        }


        @Override
        public void onNext(T t) {
            U apply = mapper.apply(t);
            actual.onNext(apply);
        }
    }
}
